package ru.ifmo.croak.impl.replication.team1.sync;


import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import ru.ifmo.croak.impl.replication.team1.LocalContext;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.net.ServerSocket;
import java.net.Socket;
import java.util.concurrent.Executor;
import java.util.concurrent.Executors;
import java.util.concurrent.atomic.AtomicBoolean;

public class SynchronizerManager implements Runnable {
    private static final Logger log = LoggerFactory.getLogger(SynchronizerManager.class);

    private final ServerSocket serverSocket;
    private final LocalContext localContext;
    private final Executor executor;

    private final AtomicBoolean canAccept;

    public SynchronizerManager(ServerSocket serverSocket, LocalContext localContext) {
        this.serverSocket = serverSocket;
        this.localContext = localContext;
        canAccept = new AtomicBoolean(true);
        executor = Executors.newCachedThreadPool();
    }

    public void synchronizeTo(final InetSocketAddress inetSocketAddress) {
        log.debug("I want synchronize with {}", inetSocketAddress);
        if (canAccept.compareAndSet(true, false)) {
            executor.execute(new Runnable() {
                public void run() {
                    new SynchronizerClient(inetSocketAddress, localContext).run();
                    canAccept.set(true);
                }
            });
        }
    }

    public void synchronizeTo(final Socket socket) {
        log.debug("I want synchronize with {}", socket);
        if (canAccept.compareAndSet(true, false)) {
            executor.execute(new Runnable() {
                public void run() {
                    new SynchronizerWorker(socket, true, localContext).run();
                    canAccept.set(true);
                }
            });
        } else {
            executor.execute(new Runnable() {
                public void run() {
                    new SynchronizerWorker(socket, false, localContext).run();
                }
            });
        }
    }


    public void run() {
        try {
            while (!Thread.interrupted()) {
                final Socket socket = serverSocket.accept();
                log.debug("accept synchronizer connection from {}", socket);
                synchronizeTo(socket);
            }
        } catch (IOException e) {
            log.error("SynchronizerManager fail", e);
        } finally {
            try {
                serverSocket.close();
            } catch (IOException e) {
                log.warn("SynchronizerManager serverSocket closing fail", e);
            }
        }
    }

}
